package operator.filter;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;
import util.Printer;
import util.RxJavaHelper;
import util.ThreadHelper;

class ThrottleOperator {

    public static void main(String[] args) {
//        throttleFirst();
//        throttleLast();
        throttleWithTimeout();
    }

    /**
     * 一订阅就开始一个窗口期，然后每次窗口期开启，取第一个onNext数据
     */
    static void throttleFirst(){
        Observable.intervalRange(1, 10, 0, 1, TimeUnit.SECONDS)
                .doAfterNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("---------------- doAfterNext aLong = " + aLong);
                    }
                })
                .throttleFirst(3, TimeUnit.SECONDS)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

    /**
     * 订阅后等待一个窗口期然后开始取数据，然后每次窗口期开启，取最后一个onNext数据
     * 功能和sample一样
     */
    static void throttleLast(){
        Observable.intervalRange(1, 10, 0, 1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("---------------- doOnNext aLong = " + aLong);
                    }
                })
                .throttleLast(3, TimeUnit.SECONDS)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

    /**
     * 功能同deboundce
     */
    static void throttleWithTimeout(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // send events with simulated time wait
                emitter.onNext(1); // skip
                Thread.sleep(400);
                emitter.onNext(2); // deliver
                Thread.sleep(505);
                emitter.onNext(3); // skip
                Thread.sleep(100);
                emitter.onNext(4); // deliver
                Thread.sleep(605);
                emitter.onNext(5); // deliver
                Thread.sleep(510);
                emitter.onComplete();
            }
        })
                .throttleWithTimeout(500, TimeUnit.MILLISECONDS)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }


}
